package simpledb;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Knows how to compute some aggregate over a set of IntFields.
 * IntegerAggregator实现一个聚合函数+groupby
 * gbIndex表示groupby的字段索引
 * aIndex表示聚合函数的字段索引
 * what执行哪一种聚合操作
 * mergeTupleIntoGroup（tuple）表示迭代一个元组的一次操作。如sum 就是加上tuple对应字段
 */


public class IntegerAggregator implements Aggregator {

    private static final long serialVersionUID = 1L;

    //指明元组哪一列被分组字段的下标
    private int gbIndex;
    //被分组字段的类型
    private Type gbIndextype;
    //指明元组哪一列被聚合字段的下标
    private int aIndex;
    //执行的聚合操作
    private Op what;
    private TupleDesc tupleDesc;
    //存储一个由  被分组索引字段->聚合的值的Map
    private Map<Field, Integer> gval2agval;
    //专门存储一个AVG操作  被分组索引字段->【数值的个数count，数值的总和sum】
    private Map<Field, Integer[]> gval2count_sum;

    /**
     * Aggregate constructor
     *
     * @param gbIndex     the 0-based index of the group-by field in the tuple, or
     *                    NO_GROUPING if there is no grouping
     * @param gbIndextype the type of the group by field (e.g., Type.INT_TYPE), or null
     *                    if there is no grouping
     * @param aIndex      the 0-based index of the aggregate field in the tuple
     * @param what        the aggregation operator
     */

    public IntegerAggregator(int gbIndex, Type gbIndextype, int aIndex, Op what) {
        this.gbIndex = gbIndex;
        this.gbIndextype = gbIndextype;
        this.aIndex = aIndex;
        this.what = what;
        this.gval2agval = new HashMap<>();
        this.gval2count_sum = new HashMap<>();
        // some code goes here
    }

    public IntegerAggregator(int gbIndex, Type gbIndextype, int aIndex, Op what, TupleDesc tupleDesc) {
        this(gbIndex, gbIndextype, aIndex, what);
        this.tupleDesc = tupleDesc;
        this.gval2agval = new HashMap<>();
        this.gval2count_sum = new HashMap<>();
    }

    /**
     * Merge a new tuple into the aggregate, grouping as indicated in the
     * constructor
     *
     *这里两个map的key要用gbField存储保证前后一致性，如果使用aggreField不同的元组会发生变化
     * 关键是使用gval2agval对不同操作进行一个更新，由于AVG包括两个数据 另外用gval2count_sum存储
     * @param tup the Tuple containing an aggregate field and a group-by field
     */
    public void mergeTupleIntoGroup(Tuple tup) {
        Field aggreField = tup.getField(aIndex);
        Field gbField = null;
        TupleDesc td = tup.getTupleDesc();
        if (gbIndex != Aggregator.NO_GROUPING) {
            gbField = tup.getField(gbIndex);
        }
        if (Type.INT_TYPE != aggreField.getType()) {
            throw new IllegalArgumentException();
        }
        int aggreField_val = ((IntField) aggreField).getValue();
        //如果是AVG操作，则使用gval2count_sum分别更新【0】count和【1】sum
        if (what == Op.AVG) {
            if (gval2count_sum.containsKey(gbField)) {
                int count = gval2count_sum.get(gbField)[0];
                int sum = gval2count_sum.get(gbField)[1];
                gval2count_sum.put(gbField, new Integer[]{count + 1, sum + aggreField_val});
            } else {
                gval2count_sum.put(gbField, new Integer[]{1, aggreField_val});
            }
            Integer[] count_sum = gval2count_sum.get(gbField);
            gval2agval.put(gbField,count_sum[1]/count_sum[0]);
        }
        //count min max sum操作
        else
        {
            int old_val = 0;
            if (!gval2agval.containsKey(gbField)) {
                old_val = initgval2agval(gbField, what);
            } else {
                old_val = gval2agval.get(gbField);
            }
            int new_val = calculateAggre(gbField, old_val,aggreField_val);
            gval2agval.put(gbField,new_val);
        }
        // some code goes here
}

    //第一个数据存储时的初始化值
    private int initgval2agval(Field gbField, Op what) {
        switch(what){
            case COUNT:
                return 0;
            case SUM:
                return 0;
            case MIN:
                return Integer.MAX_VALUE;
            case MAX:
                return Integer.MIN_VALUE;
        }
        throw new IllegalArgumentException();
    }

    //更新值变化的操作
    private int calculateAggre(Field gbField, int old_val,int aggreField_val) {
        switch(what){
            case COUNT:
                return old_val+1;
            case SUM:
                return old_val+aggreField_val;
            case MIN:
                return Math.min(old_val,aggreField_val);
            case MAX:
                return Math.max(old_val,aggreField_val);
        }
        throw new IllegalArgumentException();
    }

    /**
     * Create a OpIterator over group aggregate results.
     * 
     * @return a OpIterator whose tuples are the pair (groupVal, aggregateVal)
     *         if using group, or a single (aggregateVal) if no grouping. The
     *         aggregateVal is determined by the type of aggregate specified in
     *         the constructor.
     *         返回一个迭代器
     *         tuples中每一个元组   field1存储gbField   field2存储对应值IntField
     *         如果没有指定gbField,就直接存储对应值IntField
     */
    public OpIterator iterator() {
        List<Tuple> tuples = new ArrayList<>();
        for (Map.Entry entry :gval2agval.entrySet())
        {
           Tuple t = new Tuple(tupleDesc);
           if(gbIndex == Aggregator.NO_GROUPING)
           {
               t.setField(0,new IntField((Integer) entry.getValue()));
           }
           else
           {
               t.setField(0,(Field) entry.getKey());
               t.setField(1,new IntField((Integer) entry.getValue()));
           }
           tuples.add(t);
        }
        return new TupleIterator(tupleDesc,tuples);
    }

}
